package ahstu.dong.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.*;

import java.nio.charset.StandardCharsets;
import java.util.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class HBaseTest {
    private Configuration conf = null;
    private Connection conn = null;

    @Before
    public void init() throws Exception {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum",
                "ha01:2181,ha02:2181,ha03:2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    @Test
    public void createTable() throws Exception {
        Admin admin = conn.getAdmin();
        HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf("t_user_info".getBytes()));
        HColumnDescriptor hcd1 = new HColumnDescriptor("base_info");
        HColumnDescriptor hcd2 = new HColumnDescriptor("extra_info");
        hcd2.setVersions(1,3);
        tableDescriptor.addFamily(hcd1).addFamily(hcd2);
        admin.createTable(tableDescriptor);

        admin.close();
        conn.close();
    }

    @Test
    public void testPut() throws Exception {
        //创建table对象，通过table对象来添加数据
        Table table = conn.getTable(TableName.valueOf("t_user_info"));
        //创建一个集合，用于存放Put对象
        ArrayList<Put> puts = new ArrayList<Put>();
        //构建put对象（kv形式），并指定其行键
        Put put01 = new Put(Bytes.toBytes("user001"));
        put01.addColumn(Bytes.toBytes("base_info"),
                Bytes.toBytes("username"), Bytes.toBytes("zhangsan"));
        put01.addColumn(Bytes.toBytes("base_info"),
                Bytes.toBytes("password"), Bytes.toBytes("123456"));
        Put put02 = new Put("user002".getBytes());
        put02.addColumn(Bytes.toBytes("base_info"),
                Bytes.toBytes("username"), Bytes.toBytes("lisi"));
        put02.addColumn(Bytes.toBytes("extra_info"),
                Bytes.toBytes("married"), Bytes.toBytes("false"));
        //把所有的put对象添加到一个集合中
        puts.add(put01);
        puts.add(put02);
        //提交所有的插入数据的记录
        table.put(puts);
        //关闭
        table.close();
        conn.close();
    }

    @Test
    public void testGet() throws IOException {

        Table table = conn.getTable(TableName.valueOf("t_user_info"));
        //得到用于扫描region的对象
//        Scan scan = new Scan();
        Get get = new Get("user001".getBytes());
        //使用HTable得到resultcanner实现类的对象
        Result result1 = table.get(get);
        List<Cell> cells = result1.listCells();
        for (Cell cell : cells) {
            //得到rowkey
            System.out.println("行键:" + Bytes.toString(CellUtil.cloneRow(cell)));
            //得到列族
            System.out.println("列族:" + Bytes.toString(CellUtil.cloneFamily(cell)));
            System.out.println("列:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
            System.out.println("值:" + Bytes.toString(CellUtil.cloneValue(cell)));
        }
    }

    @Test
    public void testScan() throws Exception {
        //获取table对象
        Table table = conn.getTable(TableName.valueOf("t_user_info"));
        //获取scan对象
        Scan scan = new Scan();
        //获取查询的数据
        ResultScanner scanner = table.getScanner(scan);
        //获取ResultScanner所有数据，返回迭代器
        Iterator<Result> iter = scanner.iterator();
        //遍历迭代器
        while (iter.hasNext()) {
            //获取当前每一行结果数据
            Result result = iter.next();
            //获取当前每一行中所有的cell对象
            List<Cell> cells = result.listCells();
            //迭代所有的cell
            for(Cell c:cells){
                //获取行键
                byte[] rowArray = c.getRowArray();
                //获取列族
                byte[] familyArray = c.getFamilyArray();
                //获取列族下的列名称
                byte[] qualifierArray = c.getQualifierArray();
                //列字段的值
                byte[] valueArray = c.getValueArray();
                //打印rowArray、familyArray、qualifierArray、valueArray
                System.out.println("行键:"+new String(rowArray, c.getRowOffset(),
                        c.getRowLength()));
                System.out.print("列族:"+new String(familyArray,c.getFamilyOffset(),
                        c.getFamilyLength()));
                System.out.print(" "+"列:"+ new String(qualifierArray,
                        c.getQualifierOffset(), c.getQualifierLength()));
                System.out.println(" "+"值:"+ new String(valueArray,
                        c.getValueOffset(), c.getValueLength()));
            }
            System.out.println("-----------------------");
        }
        //关闭
        table.close();
        conn.close();
    }

    @Test
    public void testDel() throws Exception {
        //获取table对象
        Table table = conn.getTable(TableName.valueOf("t_user_info"));
        //获取delete对象,需要一个rowkey
        Delete delete = new Delete("user001".getBytes());
        //在delete对象中指定要删除的列族-列名称
        delete.addColumn("base_info".getBytes(), "password".getBytes());
        //执行删除操作
        table.delete(delete);
        //关闭
        table.close();
        conn.close();
    }

    @Test
    public void testDrop() throws Exception {
        //获取一个表的管理器
        Admin admin = conn.getAdmin();
        //删除表时先需要disable，将表置为不可用，然后在delete
        admin.disableTable(TableName.valueOf("t_user_info"));
        admin.deleteTable(TableName.valueOf("t_user_info"));
        //关闭
        admin.close();
        conn.close();
    }
}
